package xuecheng.tool.logservice.alibaba;

import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

import javax.annotation.Nonnull;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;

import com.aliyun.openservices.log.Client;
import com.aliyun.openservices.log.common.LogItem;
import com.aliyun.openservices.log.common.QueriedLog;
import com.aliyun.openservices.log.exception.LogException;
import com.aliyun.openservices.log.request.GetLogsRequest;
import com.aliyun.openservices.log.response.GetLogsResponse;
import com.google.common.collect.Lists;
import com.google.gson.FieldNamingPolicy;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;

import cn.hutool.core.util.StrUtil;
import cn.hutool.system.SystemUtil;
import lombok.extern.slf4j.Slf4j;

/**
 * 阿里日志服务工具类
 *
 * @author PengXueCheng
 * @version 1.0.0
 **/
@Slf4j
public class AliLogUtils {

    private final Client client;
    private final Gson gson;
    private final String pattern;
    private final String LINE_SEPARATOR = SystemUtil.getOsInfo().getLineSeparator();
    /**
     * 批处理最大个数
     */
    private final int MAX_BATCH_COUNT = 1000;

    public AliLogUtils(Client client) {
        this.client = client;
        this.pattern = "yyyy-MM-dd HH:mm:ss";
        this.gson = new GsonBuilder().disableHtmlEscaping().setDateFormat(pattern)
            // 与Gson一起使用此命名策略会将Java字段名称从其驼峰*格式修改为小写的字段名称，其中每个单词都用下划线（_）分隔。
            .setFieldNamingPolicy(FieldNamingPolicy.LOWER_CASE_WITH_UNDERSCORES).create();
    }

    /**
     * 分析
     *
     * @param <T>
     *            分析结果集行数据映射实体类型
     * @param project
     *            项目名称
     * @param logstore
     *            日志库名称
     * @param fromTime
     *            begin time
     * @param toTime
     *            end time
     * @param query
     *            user query(查询语句不能有换行，有换行有概率被nginx 拦截)
     * @param tClass
     *            分析结果集行数据映射实体字节码
     * @param callBack
     *            批处理回调
     */
    public <T> void analyze(String project, String logstore, Date fromTime, Date toTime, String query, Class<T> tClass,
        CallBack callBack) {
        analyze(project, logstore, fromTime, toTime, null, query, tClass, callBack);
    }

    /**
     * 分析
     *
     * @param <T>
     *            分析结果集行数据映射实体类型
     * @param project
     *            项目名称
     * @param logstore
     *            日志库名称
     * @param fromTime
     *            begin time
     * @param toTime
     *            end time
     * @param topic
     *            topic name of a log store
     * @param query
     *            user query
     * @param tClass
     *            分析结果集行数据映射实体字节码
     * @param callBack
     *            批处理回调
     */
    public <T> void analyze(String project, String logstore, Date fromTime, Date toTime, String topic, String query,
        Class<T> tClass, CallBack callBack) {
        query = dealQuery(query);

        int from = Math.toIntExact(fromTime.getTime() / 1000);
        int to = Math.toIntExact(toTime.getTime() / 1000);

        List<T> list = Lists.newArrayList();

        int logOffset = 0;
        // logLine 最大值为100，每次获取100行数据。若需要读取更多数据，请使用offset分页。offset和lines只对关键字查询有效，若使用SQL查询，则无效。在SQL查询中返回更多数据，请使用limit语法。
        int logLine = 100;
        while (true) {
            String limitStr = getLimit(logOffset, logLine);
            String tmpQuery = query + limitStr;
            GetLogsResponse resp = getResp(project, logstore, fromTime, toTime, topic, tmpQuery, from, to);
            if (resp == null) {
                // 这里也要处理数据，确保最后遗留的数据也存储到数据库
                if (callBack != null) {
                    callBack.invoke(Lists.newArrayList(list));
                    // 存储完成清理 list
                    list.clear();
                }
                break;
            }
            if (resp.IsCompleted() && resp.GetCount() == 0) {
                // 这里也要处理数据，确保最后遗留的数据也存储到数据库
                if (callBack != null) {
                    callBack.invoke(Lists.newArrayList(list));
                    // 存储完成清理 list
                    list.clear();
                }
                break;
            }
            list.addAll(convert2Entities(tClass, resp.GetLogs()));
            if (callBack != null) {
                // 达到BATCH_COUNT了，需要去处理一次，防止数据几万条数据在内存，容易OOM
                if (list.size() >= MAX_BATCH_COUNT) {
                    callBack.invoke(Lists.newArrayList(list));
                    // 存储完成清理 list
                    list.clear();
                }
            }
            logOffset += logLine;
        }
    }

    private GetLogsResponse getResp(String project, String logstore, Date fromTime, Date toTime, String topic,
        String query, int from, int to) {
        // 对于每个 log offset,一次读取 100 行 log，如果读取失败，最多重复读取 3 次。
        final int maxRetryTime = 3;
        GetLogsResponse resp = null;
        for (int retryTime = 0; retryTime < maxRetryTime; retryTime++) {
            GetLogsRequest req = new GetLogsRequest(project, logstore, from, to, topic, query);
            log.info("阿里云日志分析参数列表: project = {}, logstore = {}, fromTime = {}, toTime = {}, topic = {}, query = {}{}",
                project, logstore, DateFormatUtils.format(fromTime, pattern), DateFormatUtils.format(toTime, pattern),
                topic, LINE_SEPARATOR, query);
            try {
                resp = client.GetLogs(req);
            } catch (LogException e) {
                log.error(project + "_" + logstore + "_" + fromTime + "_" + toTime + "_" + topic + "_" + query + "_"
                    + req + "_" + e.getMessage(), e);
                continue;
            }
            if (resp != null && resp.IsCompleted()) {
                break;
            }
        }
        return resp;
    }

    private <T> List<T> convert2Entities(Class<T> tClass, List<QueriedLog> queriedLogs) {
        return queriedLogs.stream().map(queriedLog -> {
            LogItem item = queriedLog.GetLogItem();
            String json = item.ToJsonString();
            if (StringUtils.isEmpty(json)) {
                log.error("json 数据为空, queriedLog = {}, item = {}", queriedLog, item);
                return null;
            }
            return gson.fromJson(json, tClass);
        }).filter(Objects::nonNull).collect(Collectors.toList());
    }

    private String getLimit(long offset, int pageSize) {
        return String.format("%sLIMIT %s, %s", " ", offset, pageSize);
    }

    /**
     * 去除查询语句的换行符
     *
     * @param query
     * @return
     */
    private String dealQuery(String query) {
        if (StringUtils.isEmpty(query)) {
            return "";
        }

        // 去除空白字符
        query = StrUtil.trimToEmpty(query);
        // 将多个空白字符替换为一个一个空格
        query = query.replaceAll("\\s+", " ");
        return query;
    }

    public interface CallBack<T> {

        /**
         * 每次分页查询后的触发的批处理执行动作
         *
         * @param list
         *            批处理数据
         */
        void invoke(@Nonnull List<T> list);

    }

}
